﻿using Aliyun.OSS;
using Digitalmes.Configuration;
using Microsoft.Extensions.DependencyInjection;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Options;
using MQTTnet;
using MQTTnet.Packets;
using MQTTnet.Server;
using Namotion.Reflection;
using System;
using System.Security.Cryptography.X509Certificates;
using System.Text;
using System.Text.Json;
using System.Threading;
using WalkingTec.Mvvm.Core;

namespace Digitalmes.Services.Hosted
{
    public class MqttController:IMQTT
    {



     
            private readonly ILogService _logger;
            private readonly IServiceProvider _serviceProvider;
            private MqttServer _mqttServer;
            private MqttServerOptions _serverOptions;
            private DateTime _startTime = DateTime.Now;
            private readonly MqttSettings _mqttSettings;
           private readonly BatchDataService _batchDataService;

        // 存储客户端状态和消息的字典
           private readonly ConcurrentDictionary<string, ClientInfo> _connectedClients = new();
            private readonly ConcurrentDictionary<string, List<MqttMessage>> _messageStore = new();


           // 存储服务器订阅的主题和对应的处理委托
           private readonly ConcurrentDictionary<string, Func<MqttMessage, Task>> _serverSubscriptions = new();

        public MqttController(
                ILogService logger,
                IServiceProvider serviceProvider,
                BatchDataService batchDataService,
                IOptions<MqttSettings> mqttSettings,MqttServer mqtt)
            {
                _logger = logger;
                _serviceProvider = serviceProvider;
                _mqttServer=mqtt;
                _mqttSettings = mqttSettings.Value;
              _batchDataService = batchDataService;
        }

           

            /// <summary>
            /// 验证客户端连接
            /// </summary>
            /// <summary>
            /// 验证客户端连接
            /// </summary>
            public async Task ValidateConnection(ValidatingConnectionEventArgs context)
            {
                try
                {
                    var clientId = context.ClientId;
                    var username = context.UserName;

                    _logger.Information($"客户端尝试连接: {clientId}, 用户名: {username}");

                    // 检查客户端ID是否有效
                    if (string.IsNullOrEmpty(clientId) || clientId.Length > 100)
                    {
                        context.ReasonCode = MQTTnet.Protocol.MqttConnectReasonCode.ClientIdentifierNotValid;
                        _logger.Warning($"客户端ID无效: {clientId}");
                        return;
                    }

                    // 检查最大连接数限制
                    if (_connectedClients.Count >= _mqttSettings.MaxConnectedClients)
                    {
                        context.ReasonCode = MQTTnet.Protocol.MqttConnectReasonCode.QuotaExceeded;
                        _logger.Warning($"达到最大连接数限制: {_mqttSettings.MaxConnectedClients}");
                        return;
                    }

                    // 验证用户名和密码（如果有）
                    if (!string.IsNullOrEmpty(_mqttSettings.Username))
                    {
                        if (username != _mqttSettings.Username || context.Password != _mqttSettings.Password)
                        {
                            context.ReasonCode = MQTTnet.Protocol.MqttConnectReasonCode.BadUserNameOrPassword;
                            _logger.Warning($"客户端认证失败: {clientId}");
                            return;
                        }
                    }

                    // 允许连接
                    context.ReasonCode = MQTTnet.Protocol.MqttConnectReasonCode.Success;
                    _logger.Information($"客户端认证成功: {clientId}");
                }
                catch (Exception ex)
                {
                    _logger.Error("验证客户端连接时发生错误", ex);
                    context.ReasonCode = MQTTnet.Protocol.MqttConnectReasonCode.UnspecifiedError;
                }

                await Task.CompletedTask;
            }

            /// <summary>
            /// 处理客户端连接事件
            /// </summary>
            public Task HandleClientConnected(ClientConnectedEventArgs eventArgs)
            {
                var clientId = eventArgs.ClientId;

                // 记录客户端信息
                _connectedClients[clientId] = new ClientInfo
                {
                    ClientId = clientId,
                    ConnectedTime = DateTime.UtcNow,
                    LastActivity = DateTime.UtcNow
                };

                _logger.Information($"客户端已连接: {clientId}, 当前连接数: {_connectedClients.Count}");

                // 发送欢迎消息
                PublishMessage("server/welcome", $"Client {clientId} connected at {DateTime.UtcNow}");

                return Task.CompletedTask;
            }

            /// <summary>
            /// 处理客户端断开连接事件
            /// </summary>
            public Task HandleClientDisconnected(ClientDisconnectedEventArgs eventArgs)
            {
                var clientId = eventArgs.ClientId;

                if (_connectedClients.TryRemove(clientId, out _))
                {
                    _logger.Information($"客户端已断开: {clientId}, 原因: {eventArgs.ReasonString}, 当前连接数: {_connectedClients.Count}");

                    // 发布客户端断开连接的消息
                    PublishMessage("server/status", $"Client {clientId} disconnected at {DateTime.UtcNow}");
                }

                return Task.CompletedTask;
            }

            /// <summary>
            /// 验证订阅请求
            /// </summary>
            public Task ValidateSubscription(InterceptingSubscriptionEventArgs context)
            {
                var clientId = context.ClientId;
                var topic = context.TopicFilter.Topic;

                // 更新客户端活动时间
                if (_connectedClients.TryGetValue(clientId, out var clientInfo))
                {
                    clientInfo.LastActivity = DateTime.UtcNow;
                }

                _logger.Information($"客户端订阅: {clientId} -> {topic}");

                // 可以在这里添加订阅权限验证逻辑
                // 例如，检查客户端是否有权订阅特定主题

                context.ProcessSubscription = true;
                return Task.CompletedTask;
            }

            /// <summary>
            /// 拦截和处理应用消息
            /// </summary>
            public async Task InterceptApplicationMessage(InterceptingPublishEventArgs context)
            {
                var clientId = context.ClientId;
                var message = context.ApplicationMessage;
                var topic = message.Topic;

                // 使用 Payload 属性
                string payload = message.Payload.Length > 0 ?
                    Encoding.UTF8.GetString(message.Payload) :
                    string.Empty;

                // 更新客户端活动时间
                if (_connectedClients.TryGetValue(clientId, out var clientInfo))
                {
                    clientInfo.LastActivity = DateTime.UtcNow;
                    clientInfo.MessagesSent++;
                }

                _logger.Information($"收到消息: {topic} from {clientId}, 长度: {payload.Length} bytes");

                try
                {
                    // 存储消息
                    var mqttMessage = new MqttMessage
                    {
                        ClientId = clientId,
                        Topic = topic,
                        Payload = payload,
                        Timestamp = DateTime.UtcNow,
                        Qos = message.QualityOfServiceLevel
                    };

                    // 按主题存储消息
                    if (!_messageStore.ContainsKey(topic))
                    {
                        _messageStore[topic] = new List<MqttMessage>();
                    }

                    _messageStore[topic].Add(mqttMessage);


                // 处理服务器订阅的消息
                await ProcessServerSubscriptions(clientId, topic, payload, message.QualityOfServiceLevel);

                // 处理特定主题的消息
                if (topic.StartsWith("sensor/"))
                    {
                        await ProcessSensorData(clientId, topic, payload);
                    }
                    else if (topic.StartsWith("device/"))
                    {
                        await ProcessDeviceMessage(clientId, topic, payload);
                    }

                    // 可以在这里添加消息转发、持久化等逻辑
                }
                catch (Exception ex)
                {
                    _logger.Error($"处理消息时发生错误: {topic} from {clientId}", ex);
                }

                context.ProcessPublish = true;
            }

            /// <summary>
            /// 处理传感器数据
            /// </summary>
            public async Task ProcessSensorData(string clientId, string topic, string payload)
            {
                try
                {
                    // 解析传感器数据（假设是JSON格式）
                    var sensorData = JsonSerializer.Deserialize<SensorData>(payload);

                    if (sensorData != null)
                    {

                        await _batchDataService.EnqueueDataAsync(sensorData);

                       _logger.Information($"收到传感器数据: {clientId}, 类型: {sensorData.Type}, 值: {sensorData.Value}");

                        // 处理数据（例如存储到数据库）
                        using var scope = _serviceProvider.CreateScope();
                        var dbContext = scope.ServiceProvider.GetRequiredService<WTMContext>();

                        // 这里添加数据库存储逻辑
                        // await dbContext.SensorReadings.AddAsync(...);
                        // await dbContext.SaveChangesAsync();

                        // 发布处理后的数据到其他主题
                        var processedTopic = $"processed/{topic}";
                        var processedPayload = JsonSerializer.Serialize(new
                        {
                            sensorData.Type,
                            sensorData.Value,
                            ProcessedAt = DateTime.UtcNow,
                            ClientId = clientId
                        });

                        await PublishMessageAsync(processedTopic, processedPayload);
                    }
                }
                catch (Exception ex)
                {
                    _logger.Error($"处理传感器数据时发生错误: {topic} from {clientId}");
                }
            }

            /// <summary>
            /// 处理设备消息
            /// </summary>
            private async Task ProcessDeviceMessage(string clientId, string topic, string payload)
            {
                _logger.Information($"收到设备消息: {clientId}, 主题: {topic}");

                // 根据设备消息类型进行处理
                if (topic.EndsWith("/status"))
                {
                    // 处理设备状态消息
                    await PublishMessageAsync("server/device/status",
                        $"Device {clientId} status updated: {payload}");
                }
                else if (topic.EndsWith("/command"))
                {
                    // 处理设备命令
                    await ProcessDeviceCommand(clientId, payload);
                }
            }


        /// <summary>
        /// 服务器订阅主题
        /// </summary>
        /// <param name="topic">要订阅的主题</param>
        /// <param name="handler">消息处理委托</param>
        /// <param name="qos">服务质量等级</param>
        public  void Subscribe(string topic, Func<MqttMessage, Task> handler,
            MQTTnet.Protocol.MqttQualityOfServiceLevel qos = MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce)
        {
            try
            {
                // 创建主题过滤器
                var topicFilter = new MqttTopicFilter
                {
                    Topic = topic,
                    QualityOfServiceLevel = qos
                };
               
                // 存储处理委托
                _serverSubscriptions[topic] = handler;

                _logger.Information($"服务器已订阅主题: {topic}, QoS: {qos}");
                 
            }
            catch (Exception ex)
            {
                _logger.Error($"服务器订阅主题失败: {topic}", ex);
                
            }
        }

        /// <summary>
        /// 服务器取消订阅主题
        /// </summary>
        /// <param name="topic">要取消订阅的主题</param>
        public void Unsubscribe(string topic)
        {
            try
            {
               
                // 移除处理委托
                _serverSubscriptions.TryRemove(topic, out _);

                _logger.Information($"服务器已取消订阅主题: {topic}");
               
            }
            catch (Exception ex)
            {
                _logger.Error($"服务器取消订阅主题失败: {topic}", ex);
               
            }
        }



        /// <summary>
        /// 处理服务器订阅的消息
        /// </summary>
        private async Task ProcessServerSubscriptions(string clientId, string topic, string payload,
            MQTTnet.Protocol.MqttQualityOfServiceLevel qos)
        {
            // 查找匹配的订阅处理程序
            var matchingSubscriptions = _serverSubscriptions
                .Where(s => MqttTopicFilterComparer.Compare(topic, s.Key)== MqttTopicFilterCompareResult.IsMatch)
                .ToList();

            if (matchingSubscriptions.Count == 0)
                return;

            // 创建消息对象
            var mqttMessage = new MqttMessage
            {
                ClientId = clientId,
                Topic = topic,
                Payload = payload,
                Timestamp = DateTime.UtcNow,
                Qos = qos
            };

            // 处理所有匹配的订阅
            foreach (var subscription in matchingSubscriptions)
            {
                try
                {
                    if (subscription.Value.Target != null)
                    {
                        await subscription.Value(mqttMessage);
                        _logger.Debug($"已处理服务器订阅消息: {topic} -> {subscription.Key}");
                    }                 
                }
                catch (Exception ex)
                {
                    _logger.Error($"处理服务器订阅消息时发生错误: {topic} -> {subscription.Key}", ex);
                }
            }
        }

        /// <summary>
        /// 处理设备命令
        /// </summary>
        private async Task ProcessDeviceCommand(string clientId, string command)
            {
                _logger.Information($"处理设备命令: {clientId}, 命令: {command}");

                // 根据命令类型执行相应操作
                switch (command.ToLower())
                {
                    case "reboot":
                        // 发送重启命令响应
                        await PublishMessageAsync($"device/{clientId}/response", "REBOOT_ACK");
                        break;

                    case "status":
                        // 发送状态查询响应
                        await PublishMessageAsync($"device/{clientId}/response", "STATUS:ONLINE");
                        break;

                    default:
                        await PublishMessageAsync($"device/{clientId}/response", $"UNKNOWN_COMMAND: {command}");
                        break;
                }
            }

            /// <summary>
            /// 发布消息到指定主题（异步）
            /// </summary>
            public async Task PublishMessageAsync(string topic, string payload
                )
            {
                if (_mqttServer == null) return;

                try
                {
                   MQTTnet.Protocol.MqttQualityOfServiceLevel qos = MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce;

                    var message = new MqttApplicationMessageBuilder()
                        .WithTopic(topic)
                        .WithPayload(payload)
                        .WithQualityOfServiceLevel(qos)
                        .WithRetainFlag(false)
                        .Build();

                    await _mqttServer.InjectApplicationMessage(
                        new InjectedMqttApplicationMessage(message)
                        {
                            SenderClientId = "server"
                        });

                    _logger.Information($"已发布消息到主题: {topic}, 长度: {payload.Length} bytes");
                }
                catch (Exception ex)
                {
                    _logger.Error($"发布消息到主题 {topic} 时发生错误");
                }
            }

            /// <summary>
            /// 发布消息到指定主题（同步/后台）
            /// </summary>
            public  void PublishMessage(string topic, string payload)               
            {
            MQTTnet.Protocol.MqttQualityOfServiceLevel qos = MQTTnet.Protocol.MqttQualityOfServiceLevel.AtMostOnce;
                // 使用Task.Run避免阻塞调用线程
                Task.Run(() => PublishMessageAsync(topic, payload));
            }

            /// <summary>
            /// 清理过期消息
            /// </summary>
            private void CleanupOldMessages()
            {
                var cutoffTime = DateTime.UtcNow.AddHours(-1); // 保留1小时内的消息

                foreach (var topic in _messageStore.Keys.ToList())
                {
                    if (_messageStore.TryGetValue(topic, out var messages))
                    {
                        var newMessages = messages.Where(m => m.Timestamp > cutoffTime).ToList();
                        _messageStore[topic] = newMessages;
                    }
                }

                _logger.Information($"已清理过期消息，保留时间: {cutoffTime}");
            }


            /// <summary>
            /// 加载证书
            /// </summary>
            private X509Certificate2 LoadCertificate(string certificatePath, string certificatePassword)
            {
                try
                {
                    if (string.IsNullOrEmpty(certificatePath))
                        return null;

                    var cert = new X509Certificate2(
                        certificatePath,
                        certificatePassword);

                    _logger.Information($"已加载TLS证书: {cert.Subject}");
                    return cert;
                }
                catch (Exception ex)
                {
                    _logger.Error("加载TLS证书时发生错误", ex);
                    return null;
                }
            }

            /// <summary>
            /// 获取服务器统计信息
            /// </summary>
            public MqttServerStats GetStats()
            {
                return new MqttServerStats
                {
                    ConnectedClients = _connectedClients.Count,
                    TotalMessagesStored = _messageStore.Sum(t => t.Value.Count),
                    TopicsCount = _messageStore.Keys.Count,
                    Uptime = DateTime.UtcNow - _startTime,
                    ClientDetails = _connectedClients.Values.ToList()
                };
            }

            /// <summary>
            /// 获取指定主题的消息
            /// </summary>
            public List<MqttMessage> GetMessagesByTopic(string topic, int limit = 100)
            {
                if (_messageStore.TryGetValue(topic, out var messages))
                {
                    return messages
                        .OrderByDescending(m => m.Timestamp)
                        .Take(limit)
                        .ToList();
                }

                return new List<MqttMessage>();
            }


        }


   
}
